﻿// ThreadPool.cpp : 线程池方法实现
//

#include "ThreadPool.h"

const size_t TASK_MAX_THRESHOLD = 1024;
const size_t THREAD_MAX_THRESHOLD = 8;
const size_t THREAD_MAX_IDLE_TIME = 10;

ThreadPool::ThreadPool()
	: initThreadSize_(0), tasksSize_(0), idleThreadSize_(0), threadSizeThreshHold_(0), tasksQueMaxThreshHold_(TASK_MAX_THRESHOLD), poolMode_(PoolMode::MODE_FIXED), isPoolRunning_(false)
{
}

ThreadPool::~ThreadPool()
{
	isPoolRunning_ = false;

	/* 等待线程池中的线程结束执行
	 * 1. 阻塞
	 * 2. 正在执行中
	 */
	{
		std::unique_lock<std::mutex> lock(tasksQueMtx_);
		notEmpty_.notify_all(); // 唤醒阻塞线程
		exitCond_.wait(lock, [&]() -> bool
					   { return threads_.empty(); });
	}
}

bool ThreadPool::checkPoolState() const
{
	return isPoolRunning_;
}

void ThreadPool::setMode(PoolMode mode)
{
	/* 线程池运行时无法修改 */
	if (checkPoolState())
	{
		std::cerr << "mode set failed" << std::endl;
		return;
	}
	poolMode_ = mode;
}

void ThreadPool::setTaskQueMaxThreshHold(size_t threshHold)
{
	tasksQueMaxThreshHold_ = threshHold;
}

void ThreadPool::setThreadSizeThreshHold(size_t threshHold)
{
	/* 线程池运行时无法修改 */
	if (checkPoolState())
	{
		std::cerr << "thread size set failed" << std::endl;
		return;
	}

	if (poolMode_ == PoolMode::MODE_FIXED)
		threadSizeThreshHold_ = threshHold;
}

void ThreadPool::start(size_t initThreadSize)
{
	isPoolRunning_ = true;
	initThreadSize_ = initThreadSize;

	/* 创建线程对象 */
	for (size_t i = 0; i < initThreadSize_; i++)
	{
		/* 绑定线程函数 */
		auto ptr = std::make_unique<Thread>(
			std::bind(&ThreadPool::threadFunc, this, std::placeholders::_1));
		threads_[ptr->getThreadId()] = std::move(ptr); // unique_ptr 不允许拷贝构造
		idleThreadSize_++;
	}
}

Result ThreadPool::submitTask(std::shared_ptr<Task> sp)
{
	/* 获取锁 */
	std::unique_lock<std::mutex> lock(tasksQueMtx_);

	/* 线程通信 等待任务队列有空余
	 * 服务降级：用户提交任务，最长不能阻塞超过 15s，否则判断任务提交失败
	 */
	if (!notFull_.wait_for(lock, std::chrono::seconds(15),
						   [&]() -> bool
						   { return tasksQue_.size() < tasksQueMaxThreshHold_; }))
	{
		std::cerr << "Task queue is full, submit task failed!" << std::endl;
		return Result(sp, false);
	}

	/* 提交任务 */
	tasksQue_.emplace(sp);
	tasksSize_++;

	/* 在 notEmpty_ 上通知消费者 */
	notEmpty_.notify_all();

	/* cached 模式创建新线程 */
	if (poolMode_ == PoolMode::MODE_CACHED &&
		tasksSize_ > idleThreadSize_ &&
		threads_.size() < threadSizeThreshHold_)
	{
		auto ptr = std::make_unique<Thread>(
			std::bind(&ThreadPool::threadFunc, this, std::placeholders::_1));
		threads_[ptr->getThreadId()] = std::move(ptr); // unique_ptr 不允许拷贝构造
		idleThreadSize_++;
	}

	return Result(sp);
}

void ThreadPool::threadFunc(size_t threadId)
{
	auto lastExecutionTime = std::chrono::high_resolution_clock::now();
	std::cout << "Thread start - tid: " << std::this_thread::get_id() << std::endl;

	for (;;) // 不断从线程池获取任务
	{
		std::shared_ptr<Task> task_ptr;

		{ // 互斥访问任务队列
			/* 获取锁 */
			std::unique_lock<std::mutex> lock(tasksQueMtx_);

			while (tasksQue_.empty())
			{
				/* 回收线程 */
				if (!isPoolRunning_)
				{
					if (!threads_.empty())
						threads_.erase(threadId);
					exitCond_.notify_all();
					std::cout << "Thread exit - tid: " << std::this_thread::get_id() << std::endl;
					return;
				}

				if (poolMode_ == PoolMode::MODE_CACHED)
				{
					/* cached 模式 回收多余线程
					 * 空闲时间超过 60s
					 */
					if (std::cv_status::timeout ==
						notEmpty_.wait_for(lock, std::chrono::seconds(1)))
					{
						auto now = std::chrono::high_resolution_clock::now();
						auto dur = std::chrono::duration_cast<std::chrono::seconds>(now - lastExecutionTime);
						/* 回收当前线程 */
						if (dur.count() >= THREAD_MAX_IDLE_TIME &&
							threads_.size() > initThreadSize_)
						{
							/* 从线程容器中删除 */
							threads_.erase(threadId);
							idleThreadSize_--;
							std::cout << "Thread exit - tid: " << std::this_thread::get_id() << std::endl;
							return; // 结束当前线程运行
						}
					}
				}
				else
				{
					/* fixed 模式 等待任务队列有任务 */
					notEmpty_.wait(lock);
				}
			}
			idleThreadSize_--;

			/* 从任务队列取出一个任务 */
			task_ptr = tasksQue_.front();
			tasksQue_.pop();
			tasksSize_--;

			/* 还有剩余任务，继续通知其他线程 */
			if (!tasksQue_.empty())
				notEmpty_.notify_all();

			/* 在 notFull_ 上通知生产者 */
			notFull_.notify_all();
		}

		/* 当前线程执行该任务 */
		if (task_ptr != nullptr)
			task_ptr->exec();

		/* 线程空闲 */
		idleThreadSize_++;
		lastExecutionTime = std::chrono::high_resolution_clock::now();
	}
}